package data.platform.merge.internal.command;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import data.platform.common.domain.MetricValue;
import data.platform.common.event.MetricValueEvent;
import data.platform.common.merge.*;
import data.platform.common.service.command.KeyValueRepository;
import data.platform.common.service.command.MergeMessageService;
import data.platform.common.service.query.MetricTagQueryService;
import data.platform.common.util.DateUtil;
import data.platform.merge.internal.cache.AggregateItemCache;
import data.platform.merge.internal.cache.MergeItemCache;
import data.platform.merge.model.MergeItem;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;
import reactor.core.publisher.Flux;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * 通过缓存数据进行天归并
 * 天原始数据在store中不删除
 * 周，月，年归并使用store数据
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class DataAggregateSchedule {

    final ApplicationContext applicationContext;

    final MetricTagQueryService metricTagQueryService;

    final AggregateItemCache aggregateItemCache;

    final MergeItemCache mergeItemCache;

    final KeyValueRepository<String, byte[]> keyValueRepository;

    final MergeMessageService mergeMessageService;

    private DateTimeFormatter dayFormatter = DateTimeFormatter.ofPattern("yyyyMMdd");

    private DateTimeFormatter monthFormatter = DateTimeFormatter.ofPattern("yyyyMM");

    private ObjectMapper objectMapper;

    @PostConstruct
    public void init() {
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        //Mono.delay(Duration.ofMinutes(2)).doOnNext(i -> dataAggregate()).subscribe();
    }

    @Scheduled(cron = "${metric.merge.cron}")
    public void dataAggregate() {
        log.info("start data aggregate................");

        List<MetricValue> metricValues = new ArrayList<>();

        StopWatch stopWatch = new StopWatch();
        stopWatch.start();

        //LocalDate test = LocalDate.of(2021, 8, 14);
        //LocalDateTime now = LocalDateTime.of(test, LocalTime.now());
        LocalDateTime now = LocalDateTime.now();
        long nowTime = DateUtil.getDateOfLocalDateTime(now).getTime();
        // 遍历任务的所有时间窗口，如果结束时间小于当前时间，计算归并值，完成后，从缓存中删除值
        for (Iterator<MergeTimeWindow> iterator = aggregateItemCache.getCache().keySet().iterator(); iterator.hasNext(); ) {
            MergeTimeWindow mergeTimeWindow = iterator.next();
            if (mergeTimeWindow.getEndTime() < nowTime) {
                aggregateItemCache.getCache().get(mergeTimeWindow).forEach((arithmeticType, metricCache) -> {
                    metricCache.forEach((metric, tagIdCache) -> {
                        tagIdCache.forEach((tagId, aggregateItem) -> {
                            String key = AggregateItem.getKey(mergeTimeWindow, arithmeticType, metric, tagId);
                            try {
                                log.info("merge hit key: [{}].", key);
                                createMetricValue(metricValues, arithmeticType, metric, tagId, aggregateItem);
                                aggregateItemCache.clearDB(key);
                            } catch (Exception ex) {
                                log.error(key, ex);
                            }
                        });
                    });
                });
                iterator.remove();
            }
        }

        stopWatch.stop();

        sendMetricValuesEvent(metricValues);
        log.info("end data aggregate, metric value size is: {}, elapse time is: {} ms.", metricValues.size(), stopWatch.getTotalTimeMillis());

        mergeMessageService.sendMergeMessage("success");
    }

    private void createMetricValue(List<MetricValue> metricValues, ArithmeticType arithmeticType, String metric,
                                   int tagId, AggregateItem aggregateItem) {
        try {
            SortedMap<String, String> tag = getTag(tagId);
            String moType = tag.get("moc");
            List<MergeItem> mergeItems = mergeItemCache.findMergeItem(metric, moType, aggregateItem.getMergePeriod(), arithmeticType);
            for (MergeItem mergeItem : mergeItems) {
                Double value = 0.0;
                LocalDateTime eventTime = DateUtil.getDateTimeOfTimestamp(aggregateItem.getMergeTimeWindow().getStartTime());
                if (mergeItem.getArithmetic().equals("commonArithmetic")) {
                    value = getCommonArithmeticValue(mergeItem, (CommonAggregateItem) aggregateItem);
                } else if (mergeItem.getArithmetic().equals("maxOccurTimeArithmetic")) {
                    CommonAggregateItem commonAggregateItem = (CommonAggregateItem) aggregateItem;
                    value = commonAggregateItem.getMax();
                    eventTime = commonAggregateItem.getMaxEventTime();
                } else if (mergeItem.getArithmetic().equals("minOccurTimeArithmetic")) {
                    CommonAggregateItem commonAggregateItem = (CommonAggregateItem) aggregateItem;
                    value = commonAggregateItem.getMin();
                    eventTime = commonAggregateItem.getMinEventTime();
                } else if (mergeItem.getArithmetic().equals("lastPeriodValueArithmetic")) {
                    CommonAggregateItem commonAggregateItem = (CommonAggregateItem) aggregateItem;
                    value = commonAggregateItem.getLatestValue();
                    eventTime = commonAggregateItem.getLatestEventTime();
                } else if (mergeItem.getArithmetic().equals("highLoadPercentArithmetic")
                        || mergeItem.getArithmetic().equals("midLoadPercentArithmetic")
                        || mergeItem.getArithmetic().equals("lowLoadPercentArithmetic")) {
                    value = getLoadPercentArithmeticValue(mergeItem, (LoadPercentAggregateItem) aggregateItem);
                } else if (mergeItem.getArithmetic().equals("ifCrowdTimeArithmetic")) {
                    IfCrowdTimeAggregateItem ifCrowdTimeAggregateItem = (IfCrowdTimeAggregateItem) aggregateItem;
                    value = Double.valueOf(ifCrowdTimeAggregateItem.getCrowdCount()) / ifCrowdTimeAggregateItem.getCount();
                } else if (mergeItem.getArithmetic().equals("faultArithmetic")) {
                    value = getFaultArithmeticValue(mergeItem, (FaultAggregateItem) aggregateItem);
                }

                addTag(tag, mergeItem.getMergePeriod());
                metricValues.add(createMetricValue(mergeItem.getMergedMetric(), tag, eventTime, value));
            }
        } catch (Exception ex) {
            log.error("", ex);
        }
    }


    private void addTag(Map<String, String> tag, MergePeriod mergePeriod) {
        tag.put("reportCycle", String.valueOf(mergePeriod.getPeriod()));
        tag.put("reportDataType", String.valueOf(0));
    }

    private SortedMap<String, String> getTag(Integer tagId) throws JsonProcessingException {
        String tagJson = metricTagQueryService.getTagJson(tagId).get();
        SortedMap<String, String> tag = objectMapper.readValue(tagJson, new TypeReference<SortedMap<String, String>>() {
        });
        return tag;
    }

    private Double getCommonArithmeticValue(MergeItem mergeItem, CommonAggregateItem aggregateItem) {
        Double value;
        switch (mergeItem.getMergeType()) {
            case SUM:
                value = aggregateItem.getSum();
                break;
            case MIN:
                value = aggregateItem.getMin();
                break;
            case MAX:
                value = aggregateItem.getMax();
                break;
            default:
                value = aggregateItem.getAvg();
                break;
        }
        return value;
    }

    private Double getLoadPercentArithmeticValue(MergeItem mergeItem, LoadPercentAggregateItem aggregateItem) {
        Double value;
        switch (mergeItem.getArithmetic()) {
            case "highLoadPercentArithmetic":
                value = Double.valueOf(aggregateItem.getHighCount()) / aggregateItem.getCount();
                break;
            case "midLoadPercentArithmetic":
                value = Double.valueOf(aggregateItem.getMidCount()) / aggregateItem.getCount();
                break;
            case "lowLoadPercentArithmetic":
            default:
                value = Double.valueOf(aggregateItem.getLowCount()) / aggregateItem.getCount();
                break;
        }
        return value;
    }

    private Double getFaultArithmeticValue(MergeItem mergeItem, FaultAggregateItem aggregateItem) {
        aggregateItem.windowEndCompute();
        Double value;
        switch (mergeItem.getMergedMetric()) {
            case "faultCount":
                value = Double.valueOf(aggregateItem.getFaultCount());
                break;
            case "faultTime":
                value = Double.valueOf(aggregateItem.getFaultTime());
                break;
            case "runRate":
                value = aggregateItem.getRunRate();
                break;
            default:
                value = Double.valueOf(aggregateItem.getMtbf());
                break;
        }
        return value;
    }

    private MetricValue createMetricValue(String metric, Map<String, String> tag, LocalDateTime eventTime, Double value) {
        MetricValue metricValue = MetricValue.builder()
                .metric(metric)
                .tag(tag)
                .eventTime(eventTime)
                .value(value)
                .build();
        return metricValue;
    }

    private void sendMetricValuesEvent(List<MetricValue> metricValues) {
        // 数据通过消息发送给时序数据处理程序
        Flux.fromIterable(metricValues)
                .bufferTimeout(100, Duration.ofMillis(500))
                .subscribe(mvs -> {
                    applicationContext.publishEvent(new MetricValueEvent(mvs));
                    log.info("send metricValue event, size is:{}.", mvs.size());
                });
    }
}
